package com.etc.wordcount;



import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.List;
import java.util.Map;

/**
 * @author 上等的猪头肉
 * @Title: SplitBolt
 * @ProjectName storm
 * @Description: TODO
 * @date 2018/11/15 16:08
 */
public class SplitBolt extends BaseRichBolt {

    OutputCollector collector;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        this.collector = outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {

        String line = tuple.getString(0);
        System.err.println("**SplitBolt  line: " + line);

        String[] words = line.split(" ");
        for (String word : words) {
            List t = new Values(word, 1);
            this.collector.emit(t);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

        outputFieldsDeclarer.declare(new Fields("word","number"));
    }
}
